package com.runssnail.demo.kafka.jms;

import org.apache.commons.lang3.Validate;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;

/**
 * Kafka TopicPublisher
 * <p/>
 * Created by zhengwei on 2015/8/27.
 */
public class KafkaTopicPublisher implements TopicPublisher {

    private static final Logger log = LoggerFactory.getLogger(KafkaTopicPublisher.class);

    /**
     * KafkaProducer
     */
    private KafkaProducer kafkaProducer;

    public KafkaTopicPublisher(KafkaProducer kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public Topic getTopic() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void publish(Message message) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public void publish(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public void publish(final Topic topic, final Message message) throws JMSException {

        Validate.notNull(topic, "Topic is required");
        Validate.notNull(topic.getTopicName(), "Topic name is required");

        Validate.notNull(message, "Message is required");
        Validate.isTrue(message instanceof ObjectMessage, "the message must be a ObjectMessage");

        ObjectMessage objectMessage = ((ObjectMessage) message);
        Validate.notNull(objectMessage.getObject(), "the message's object is required");
        Validate.notNull(message.getJMSMessageID(), "message's id is required");

        if (log.isDebugEnabled()) {
            log.debug("publish message start, topic=" + topic.getTopicName() + ", message=" + message);
        }

        final String topicName = topic.getTopicName();

        final String messageKey = topicName + "-" + message.getJMSMessageID();

        // 异步发送
        this.kafkaProducer.send(new ProducerRecord<String, Message>(topicName, messageKey,
                message), new Callback() {

            @Override
            public void onCompletion(RecordMetadata metadata, Exception ex) {
                if (ex != null) {
                    log.error("send message to server success, topic=" + topicName + ", messageId=" + messageKey + ", partition=" + metadata.partition() + ", offset=" + metadata.offset(), ex);
                } else {
                    if (log.isDebugEnabled()) {
                        log.debug("send message to server success, topic=" + topicName + ", messageId=" + messageKey + ", partition=" + metadata.partition() + ", offset=" + metadata.offset());
                    }
                }

            }
        });
    }

    @Override
    public void publish(Topic topic, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public boolean getDisableMessageID() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setDisableMessageID(boolean b) throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean getDisableMessageTimestamp() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setDisableMessageTimestamp(boolean b) throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public int getDeliveryMode() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setDeliveryMode(int i) throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public int getPriority() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setPriority(int i) throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public long getTimeToLive() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setTimeToLive(long l) throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public Destination getDestination() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void close() throws JMSException {
        this.kafkaProducer.close();
    }

    @Override
    public void send(Message message) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public void send(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public void send(Destination destination, Message message) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }

    @Override
    public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        throw new UnsupportedOperationException("use publish(Topic topic, Message message)");
    }
}
